package cn.itcast.streaming.function.flatmap;

import cn.itcast.streaming.bean.ElectricFenceModel;
import cn.itcast.streaming.bean.ElectricFenceRuleTmp;
import cn.itcast.streaming.bean.ItcastDataObj;
import cn.itcast.streaming.utils.DateUtil;
import cn.itcast.streaming.utils.DistanceCaculateUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;

//自定义CoflatmapFunction对象，实现原始数据与电子围栏规则数据合并操作，返回电子围栏规则模型流数据             vin      电子围栏规则
public class ElectricFenceRulesFunction implements CoFlatMapFunction<ItcastDataObj, HashMap<String, ElectricFenceRuleTmp>, ElectricFenceModel> {
    private HashMap<String,ElectricFenceRuleTmp> electricFenceRule = new HashMap<>();

    //作用于原始数据的flatmap操作
    public void flatMap1(ItcastDataObj itcastDataObj, Collector<ElectricFenceModel> collector) throws Exception {
        ElectricFenceModel model = new ElectricFenceModel();
        //验证原始数据是否存在经纬度以及地理位置
        if(itcastDataObj.getLng()!=-999999D && itcastDataObj.getLat() != -999999D
                && itcastDataObj.getLng() != 0D && itcastDataObj.getLat()!=0D &&
                StringUtils.isNotEmpty(itcastDataObj.getGpsTime())){
            //原始数据正确，可以与电子围栏规则判断，
            String vin = itcastDataObj.getVin();
            ElectricFenceRuleTmp electricFenceRuleTmp = electricFenceRule.getOrDefault(vin, null);
            if(electricFenceRuleTmp!=null){
                //车辆被监控了
                //获取车辆GPS的时间
                long gpsTimestamp = DateUtil.convertStringToDate(itcastDataObj.getGpsTime()).getTime();
                //判断监控的车辆对应的电子围栏规则是否在生效期内
                if(electricFenceRuleTmp.getStart_time().getTime()<=gpsTimestamp &&
                        gpsTimestamp<= electricFenceRuleTmp.getEnd_time().getTime()){
                    //电子围栏规则在生效期内
                    model.setVin(itcastDataObj.getVin());
                    model.setGpsTime(itcastDataObj.getGpsTime());
                    model.setLng(itcastDataObj.getLng()); //车辆经纬度
                    model.setLat(itcastDataObj.getLat());
                    model.setTerminalTime(itcastDataObj.getTerminalTime());
                    model.setTerminalTimestamp(itcastDataObj.getTerminalTimeStamp());

                    //电子围栏数据赋值
                    model.setEleId(electricFenceRuleTmp.getId());
                    model.setEleName(electricFenceRuleTmp.getName());
                    model.setAddress(electricFenceRuleTmp.getAddress());
                    model.setRadius(electricFenceRuleTmp.getRadius()); //半径
                    model.setLongitude(electricFenceRuleTmp.getLongitude());//电子围栏中心经纬度
                    model.setLatitude(electricFenceRuleTmp.getLatitude());

                    //计算两个经纬度之间的距离
                    Double distance = DistanceCaculateUtil.getDistance(
                            itcastDataObj.getLat(), itcastDataObj.getLng(),
                            electricFenceRuleTmp.getLatitude(), electricFenceRuleTmp.getLongitude());
                    if(distance/1000<=electricFenceRuleTmp.getRadius()){
                        //在围栏内部
                        model.setNowStatus(0);
                    }else {
                        //在围栏外部
                        model.setNowStatus(1);
                    }
                    System.out.println("拉宽后的结果数据"+model);
                    collector.collect(model);
                }else {
                    System.out.println("电子围栏"+electricFenceRuleTmp.getId()+"没有或不在生效期内");
                }
            }else {
                System.out.println("vin:"+vin+"没有符合的电子围栏规则，因此不需要监控");
            }
        }
    }

    //作用于规则广播流
    @Override
    public void flatMap2(HashMap<String, ElectricFenceRuleTmp> stringElectricFenceRuleTmpHashMap,
                         Collector<ElectricFenceModel> collector) throws Exception {
        electricFenceRule = stringElectricFenceRuleTmpHashMap;
    }
}
